package common

import (
	"encoding/json"
	"fmt"
	"os"
)

// 管理reduce任务
func DoReduce(
	// 参数列表
	jobName string, // 任务名称
	reduceTaskNumber int, // reduce任务编号
	outFile string, // 输出文件
	nMap int, // 运行的map任务号
	reduceF func(key string, value []string) string) {

	var result = make(map[string][]string)

	// 打开每一个中间文件
	for i := 0; i < nMap; i++ {
		interFile := GetReduceName(jobName, i, reduceTaskNumber)
		fmt.Println("worker : open reduce file : ", interFile)
		f, err := os.Open(interFile)
		if err != nil {
			fmt.Println("error : ", err.Error())
		}

		defer f.Close()

		decoder := json.NewDecoder(f)

		var kv KeyValue

		for decoder.More() {
			err = decoder.Decode(&kv)
			if err != nil {
				fmt.Println("error : ", err.Error())
			}

			// 将具有相同key的内容进行合并
			result[kv.Key] = append(result[kv.Key], kv.Value)
		}
	}

	// 把内容做相应的处理 然后存入最终输出文件
	// 新建输出文件
	out, err := os.Create(outFile)
	if err != nil {
		fmt.Println("error : ", err.Error())
	}

	defer out.Close()

	encoder := json.NewEncoder(out)
	for key, value := range result {
		encoder.Encode(KeyValue{Key: key, Value: reduceF(key, value)})
	}
}
